package com.siteseat.crudeq.kernel;

import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.atomic.AtomicInteger;

import com.siteseat.common.exception.InternalException;
import com.siteseat.crudeq.common.raw.Constants;
import com.siteseat.crudeq.common.raw.MessageHelper;
import com.siteseat.crudeq.kernel.exception.CreateChunkFailureException;

/**
 *
 */
public class Chunk implements Constants {

  private KernelFactory factory;
  
  private ChunkInfo     info;

  private AtomicInteger preparingCounter;

  private AtomicInteger offset;

  private byte[]        payload;

  private int           lastSavePoint;

  private ChunkWriter   writer;

  /**
   * @param profile
   * @param path
   * @param cid
   * @param size
   * @return
   * @throws IOException
   * @throws CreateChunkFailureException
   * @throws IOException
   */
  public Chunk( KernelFactory factory, ChunkInfo info, int capacity ) throws IOException {
    this.factory = factory;
    this.info = info;
    this.writer = factory.newChunkWriter( info );
    this.offset = new AtomicInteger( 0 );
    this.payload = new byte[capacity];
    this.preparingCounter = new AtomicInteger( 0 );
    this.lastSavePoint = 0;
  }
  

  /**
   * Gets the factory
   *
   * @return the factory
   */
  public KernelFactory getFactory() {
    return factory;
  }


  /**
   * @return
   */
  public int computeNewSavePoint() {
    int offsetSnapshot = this.offset.get();
    if( this.preparingCounter.get() == 0 ) {
      return offsetSnapshot;
    }

    int r = this.lastSavePoint;
    byte[] raw = this.payload;
    while( MessageHelper.isCommitted( raw, r ) ) {
      r += MessageHelper.readSize( raw, r );
    }
    return r;
  }

  /**
   * @throws IOException
   */
  public void save() throws IOException {
    this.writer.save( this );
  }

  /**
   * @param ous
   * @throws IOException
   */
  public int saveTo( ChunkInfo ci, OutputStream ous ) throws IOException {
    assert ci == getInfo();

    int newSavePoint = computeNewSavePoint();
    int count = newSavePoint - this.lastSavePoint;
    if( count > 0 ) {
      ous.write( this.payload, this.lastSavePoint, count );
      this.lastSavePoint = newSavePoint;
    }
    return count;
  }

  /**
   * Gets the chunkId
   * 
   * @return the chunkId
   */
  public ChunkInfo getInfo() {
    return info;
  }

  /**
   * Lock-free for concurrency.
   * 
   * @param size
   * @return
   */
  public Message prepareMessage( int messageBodySize ) {
    if( messageBodySize < 0 ) {
      throw new InternalException( "Invalid message body size <" + messageBodySize + ">. Should be >= 0" );
    }
    int messageSize = messageBodySize + MESSAGE.HEADER_SIZE;

    do {
      // Get a snapshot of localOffset offset;
      int curr = this.offset.get();
      int newOffset = curr + messageSize;
      if( newOffset > this.payload.length ) {
        // The chunk is already full
        return null;
      }

      // Try to allocate
      if( this.offset.compareAndSet( curr, newOffset ) ) {
        for( ;; ) {
          int oldCounter = this.preparingCounter.get();
          if( this.preparingCounter.compareAndSet( oldCounter, oldCounter + 1 ) ) {
            break;
          }
        }

        MessageHelper.prepare( this.payload, curr, messageSize );        
        return getFactory().newMessage( this, curr, messageSize );
      }

      // Another thread did allocation right after we get a snapshot and before
      // we call compareAndSet()
    } while( true );
  }

  /**
   * @return
   */
  public boolean notAllSaved() {
    int offs = getOffset();
    return offs > 0 && offs > this.lastSavePoint;
  }

  /**
   * @throws IOException
   */
  public void close() throws IOException {
    this.writer.close();
  }

  /**
   * @return
   */
  public int getOffset() {
    return this.offset.get();
  }

  /**
   * 
   */
  public void submit( Message msg, boolean commit ) {
    assert msg.getChunk() == this;

    MessageHelper.setCommitted( this.payload, msg.getRawOffset(), commit );

    for( ;; ) {
      int oldCounter = this.preparingCounter.get();
      if( this.preparingCounter.compareAndSet( oldCounter, oldCounter - 1 ) ) {
        break;
      }
    }
  }

  /**
   * 
   */
  public void commit( Message msg ) {
    submit( msg, true );
  }

  public void rollback( Message msg ) {
    submit( msg, false );
  }

  /**
   * Gets the payload
   * 
   * @return the payload
   */
  public byte[] getPayload() {
    return payload;
  }

}
